// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#pragma once

#include <array>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <limits>
#include <memory>
#include <numeric>
#include <string>
#include <string_view>
#include <vector>

#include <nebula/array/array_base.h>
#include <nebula/array/array_binary.h>
#include <nebula/array/builder_base.h>
#include <nebula/array/data.h>
#include <nebula/core/buffer.h>
#include <nebula/core/buffer_builder.h>
#include <turbo/utility/status.h>
#include <nebula/types/type.h>
#include <nebula/util/binary_view_util.h>
#include <turbo/base/macros.h>

namespace nebula {

    /// \addtogroup binary-builders
    ///
    /// @{

    // ----------------------------------------------------------------------
    // Binary and String

    template<typename TYPE>
    class BaseBinaryBuilder
            : public ArrayBuilder,
              public internal::ArrayBuilderExtraOps<BaseBinaryBuilder<TYPE>, std::string_view> {
    public:
        using TypeClass = TYPE;
        using offset_type = typename TypeClass::offset_type;

        explicit BaseBinaryBuilder(MemoryPool *pool = default_memory_pool(),
                                   int64_t alignment = kDefaultBufferAlignment)
                : ArrayBuilder(pool, alignment),
                  offsets_builder_(pool, alignment),
                  value_data_builder_(pool, alignment) {}

        BaseBinaryBuilder(const std::shared_ptr<DataType> &type, MemoryPool *pool)
                : BaseBinaryBuilder(pool) {}

        turbo::Status append(const uint8_t *value, offset_type length) {
            TURBO_RETURN_NOT_OK(Reserve(1));
            UnsafeAppendNextOffset();
            // Safety check for UBSAN.
            if (TURBO_LIKELY(length > 0)) {
                TURBO_RETURN_NOT_OK(validate_overflow(length));
                TURBO_RETURN_NOT_OK(value_data_builder_.append(value, length));
            }

            unsafe_append_to_bitmap(true);
            return turbo::OkStatus();
        }

        turbo::Status append(const char *value, offset_type length) {
            return append(reinterpret_cast<const uint8_t *>(value), length);
        }

        turbo::Status append(std::string_view value) {
            return append(value.data(), static_cast<offset_type>(value.size()));
        }

        /// Extend the last appended value by appending more data at the end
        ///
        /// Unlike append, this does not create a new offset.
        turbo::Status extend_current(const uint8_t *value, offset_type length) {
            // Safety check for UBSAN.
            if (TURBO_LIKELY(length > 0)) {
                TURBO_RETURN_NOT_OK(validate_overflow(length));
                TURBO_RETURN_NOT_OK(value_data_builder_.append(value, length));
            }
            return turbo::OkStatus();
        }

        turbo::Status extend_current(std::string_view value) {
            return extend_current(reinterpret_cast<const uint8_t *>(value.data()),
                                 static_cast<offset_type>(value.size()));
        }

        turbo::Status append_nulls(int64_t length) final {
            const int64_t num_bytes = value_data_builder_.length();
            TURBO_RETURN_NOT_OK(Reserve(length));
            for (int64_t i = 0; i < length; ++i) {
                offsets_builder_.unsafe_append(static_cast<offset_type>(num_bytes));
            }
            unsafe_append_to_bitmap(length, false);
            return turbo::OkStatus();
        }

        turbo::Status append_null() final {
            TURBO_RETURN_NOT_OK(Reserve(1));
            UnsafeAppendNextOffset();
            unsafe_append_to_bitmap(false);
            return turbo::OkStatus();
        }

        turbo::Status append_empty_value() final {
            TURBO_RETURN_NOT_OK(Reserve(1));
            UnsafeAppendNextOffset();
            unsafe_append_to_bitmap(true);
            return turbo::OkStatus();
        }

        turbo::Status append_empty_values(int64_t length) final {
            const int64_t num_bytes = value_data_builder_.length();
            TURBO_RETURN_NOT_OK(Reserve(length));
            for (int64_t i = 0; i < length; ++i) {
                offsets_builder_.unsafe_append(static_cast<offset_type>(num_bytes));
            }
            unsafe_append_to_bitmap(length, true);
            return turbo::OkStatus();
        }

        /// \brief append without checking capacity
        ///
        /// Offsets and data should have been presized using Reserve() and
        /// reserve_data(), respectively.
        void unsafe_append(const uint8_t *value, offset_type length) {
            UnsafeAppendNextOffset();
            value_data_builder_.unsafe_append(value, length);
            unsafe_append_to_bitmap(true);
        }

        void unsafe_append(const char *value, offset_type length) {
            unsafe_append(reinterpret_cast<const uint8_t *>(value), length);
        }

        void unsafe_append(const std::string &value) {
            unsafe_append(value.c_str(), static_cast<offset_type>(value.size()));
        }

        void unsafe_append(std::string_view value) {
            unsafe_append(value.data(), static_cast<offset_type>(value.size()));
        }

        /// Like extend_current, but do not check capacity
        void unsafe_extend_current(const uint8_t *value, offset_type length) {
            value_data_builder_.unsafe_append(value, length);
        }

        void unsafe_extend_current(std::string_view value) {
            unsafe_extend_current(reinterpret_cast<const uint8_t *>(value.data()),
                                static_cast<offset_type>(value.size()));
        }

        void unsafe_append_null() {
            const int64_t num_bytes = value_data_builder_.length();
            offsets_builder_.unsafe_append(static_cast<offset_type>(num_bytes));
            unsafe_append_to_bitmap(false);
        }

        void unsafe_append_empty_value() {
            const int64_t num_bytes = value_data_builder_.length();
            offsets_builder_.unsafe_append(static_cast<offset_type>(num_bytes));
            unsafe_append_to_bitmap(true);
        }

        /// \brief append a sequence of strings in one shot.
        ///
        /// \param[in] values a vector of strings
        /// \param[in] valid_bytes an optional sequence of bytes where non-zero
        /// indicates a valid (non-null) value
        /// \return turbo::Status
        turbo::Status append_values(const std::vector<std::string> &values,
                                   const uint8_t *valid_bytes = nullptr) {
            std::size_t total_length = std::accumulate(
                    values.begin(), values.end(), 0ULL,
                    [](uint64_t sum, const std::string &str) { return sum + str.size(); });
            TURBO_RETURN_NOT_OK(Reserve(values.size()));
            TURBO_RETURN_NOT_OK(reserve_data(total_length));

            if (valid_bytes != nullptr) {
                for (std::size_t i = 0; i < values.size(); ++i) {
                    UnsafeAppendNextOffset();
                    if (valid_bytes[i]) {
                        value_data_builder_.unsafe_append(
                                reinterpret_cast<const uint8_t *>(values[i].data()), values[i].size());
                    }
                }
            } else {
                for (const auto &value: values) {
                    UnsafeAppendNextOffset();
                    value_data_builder_.unsafe_append(reinterpret_cast<const uint8_t *>(value.data()),
                                                     value.size());
                }
            }

            unsafe_append_to_bitmap(valid_bytes, values.size());
            return turbo::OkStatus();
        }

        /// \brief append a sequence of nul-terminated strings in one shot.
        ///        If one of the values is NULL, it is processed as a null
        ///        value even if the corresponding valid_bytes entry is 1.
        ///
        /// \param[in] values a contiguous C array of nul-terminated char *
        /// \param[in] length the number of values to append
        /// \param[in] valid_bytes an optional sequence of bytes where non-zero
        /// indicates a valid (non-null) value
        /// \return turbo::Status
        turbo::Status append_values(const char **values, int64_t length,
                                   const uint8_t *valid_bytes = nullptr) {
            std::size_t total_length = 0;
            std::vector<std::size_t> value_lengths(length);
            bool have_null_value = false;
            for (int64_t i = 0; i < length; ++i) {
                if (values[i] != nullptr) {
                    auto value_length = strlen(values[i]);
                    value_lengths[i] = value_length;
                    total_length += value_length;
                } else {
                    have_null_value = true;
                }
            }
            TURBO_RETURN_NOT_OK(Reserve(length));
            TURBO_RETURN_NOT_OK(reserve_data(total_length));

            if (valid_bytes) {
                int64_t valid_bytes_offset = 0;
                for (int64_t i = 0; i < length; ++i) {
                    UnsafeAppendNextOffset();
                    if (valid_bytes[i]) {
                        if (values[i]) {
                            value_data_builder_.unsafe_append(reinterpret_cast<const uint8_t *>(values[i]),
                                                             value_lengths[i]);
                        } else {
                            unsafe_append_to_bitmap(valid_bytes + valid_bytes_offset,
                                                 i - valid_bytes_offset);
                            unsafe_append_to_bitmap(false);
                            valid_bytes_offset = i + 1;
                        }
                    }
                }
                unsafe_append_to_bitmap(valid_bytes + valid_bytes_offset, length - valid_bytes_offset);
            } else {
                if (have_null_value) {
                    std::vector<uint8_t> valid_vector(length, 0);
                    for (int64_t i = 0; i < length; ++i) {
                        UnsafeAppendNextOffset();
                        if (values[i]) {
                            value_data_builder_.unsafe_append(reinterpret_cast<const uint8_t *>(values[i]),
                                                             value_lengths[i]);
                            valid_vector[i] = 1;
                        }
                    }
                    unsafe_append_to_bitmap(valid_vector.data(), length);
                } else {
                    for (int64_t i = 0; i < length; ++i) {
                        UnsafeAppendNextOffset();
                        value_data_builder_.unsafe_append(reinterpret_cast<const uint8_t *>(values[i]),
                                                         value_lengths[i]);
                    }
                    unsafe_append_to_bitmap(nullptr, length);
                }
            }
            return turbo::OkStatus();
        }

        turbo::Status append_array_slice(const ArraySpan &array, int64_t offset,
                                       int64_t length) override {
            auto bitmap = array.get_values<uint8_t>(0, 0);
            auto offsets = array.get_values<offset_type>(1);
            auto data = array.get_values<uint8_t>(2, 0);
            auto total_length = offsets[offset + length] - offsets[offset];
            TURBO_RETURN_NOT_OK(Reserve(length));
            TURBO_RETURN_NOT_OK(reserve_data(total_length));
            for (int64_t i = 0; i < length; i++) {
                if (!bitmap || bit_util::get_bit(bitmap, array.offset + offset + i)) {
                    const offset_type start = offsets[offset + i];
                    const offset_type end = offsets[offset + i + 1];
                    unsafe_append(data + start, end - start);
                } else {
                    unsafe_append_null();
                }
            }
            return turbo::OkStatus();
        }

        void reset() override {
            ArrayBuilder::reset();
            offsets_builder_.reset();
            value_data_builder_.reset();
        }

        turbo::Status validate_overflow(int64_t new_bytes) {
            auto new_size = value_data_builder_.length() + new_bytes;
            if (TURBO_UNLIKELY(new_size > memory_limit())) {
                return turbo::resource_exhausted_error("array cannot contain more than ", memory_limit(),
                                                       " bytes, have ", new_size);
            } else {
                return turbo::OkStatus();
            }
        }

        turbo::Status resize(int64_t capacity) override {
            TURBO_RETURN_NOT_OK(check_capacity(capacity));
            // One more than requested for offsets
            TURBO_RETURN_NOT_OK(offsets_builder_.resize(capacity + 1));
            return ArrayBuilder::resize(capacity);
        }

        /// \brief Ensures there is enough allocated capacity to append the indicated
        /// number of bytes to the value data buffer without additional allocations
        turbo::Status reserve_data(int64_t elements) {
            TURBO_RETURN_NOT_OK(validate_overflow(elements));
            return value_data_builder_.Reserve(elements);
        }

        turbo::Status finish_internal(std::shared_ptr<ArrayData> *out) override {
            // write final offset (values length)
            TURBO_RETURN_NOT_OK(AppendNextOffset());

            // These buffers' padding zeroed by BufferBuilder
            std::shared_ptr<Buffer> offsets, value_data, null_bitmap;
            TURBO_RETURN_NOT_OK(offsets_builder_.finish(&offsets));
            TURBO_RETURN_NOT_OK(value_data_builder_.finish(&value_data));
            TURBO_RETURN_NOT_OK(null_bitmap_builder_.finish(&null_bitmap));

            *out = ArrayData::create(type(), length_, {null_bitmap, offsets, value_data},
                                     null_count_, 0);
            reset();
            return turbo::OkStatus();
        }

        /// \return data pointer of the value date builder
        const uint8_t *value_data() const { return value_data_builder_.data(); }

        /// \return size of values buffer so far
        int64_t value_data_length() const { return value_data_builder_.length(); }

        /// \return capacity of values buffer
        int64_t value_data_capacity() const { return value_data_builder_.capacity(); }

        /// \return data pointer of the value date builder
        const offset_type *offsets_data() const { return offsets_builder_.data(); }

        /// Temporary access to a value.
        ///
        /// This pointer becomes invalid on the next modifying operation.
        const uint8_t *get_value(int64_t i, offset_type *out_length) const {
            const offset_type *offsets = offsets_builder_.data();
            const auto offset = offsets[i];
            if (i == (length_ - 1)) {
                *out_length = static_cast<offset_type>(value_data_builder_.length()) - offset;
            } else {
                *out_length = offsets[i + 1] - offset;
            }
            return value_data_builder_.data() + offset;
        }

        offset_type offset(int64_t i) const { return offsets_data()[i]; }

        /// Temporary access to a value.
        ///
        /// This view becomes invalid on the next modifying operation.
        std::string_view get_view(int64_t i) const {
            offset_type value_length;
            const uint8_t *value_data = get_value(i, &value_length);
            return std::string_view(reinterpret_cast<const char *>(value_data), value_length);
        }

        // Cannot make this a static attribute because of linking issues
        static constexpr int64_t memory_limit() {
            return std::numeric_limits<offset_type>::max() - 1;
        }

    protected:
        TypedBufferBuilder<offset_type> offsets_builder_;
        TypedBufferBuilder<uint8_t> value_data_builder_;

        turbo::Status AppendNextOffset() {
            const int64_t num_bytes = value_data_builder_.length();
            return offsets_builder_.append(static_cast<offset_type>(num_bytes));
        }

        void UnsafeAppendNextOffset() {
            const int64_t num_bytes = value_data_builder_.length();
            offsets_builder_.unsafe_append(static_cast<offset_type>(num_bytes));
        }
    };

    /// \class BinaryBuilder
    /// \brief Builder class for variable-length binary data
    class TURBO_EXPORT BinaryBuilder : public BaseBinaryBuilder<BinaryType> {
    public:
        using BaseBinaryBuilder::BaseBinaryBuilder;

        /// \cond FALSE
        using ArrayBuilder::finish;
        /// \endcond

        turbo::Status finish(std::shared_ptr<BinaryArray> *out) { return finish_typed(out); }

        std::shared_ptr<DataType> type() const override { return binary(); }
    };

    /// \class StringBuilder
    /// \brief Builder class for UTF8 strings
    class TURBO_EXPORT StringBuilder : public BinaryBuilder {
    public:
        using BinaryBuilder::BinaryBuilder;

        /// \cond FALSE
        using ArrayBuilder::finish;
        /// \endcond

        turbo::Status finish(std::shared_ptr<StringArray> *out) { return finish_typed(out); }

        std::shared_ptr<DataType> type() const override { return utf8(); }
    };

    /// \class LargeBinaryBuilder
    /// \brief Builder class for large variable-length binary data
    class TURBO_EXPORT LargeBinaryBuilder : public BaseBinaryBuilder<LargeBinaryType> {
    public:
        using BaseBinaryBuilder::BaseBinaryBuilder;

        /// \cond FALSE
        using ArrayBuilder::finish;
        /// \endcond

        turbo::Status finish(std::shared_ptr<LargeBinaryArray> *out) { return finish_typed(out); }

        std::shared_ptr<DataType> type() const override { return large_binary(); }
    };

    /// \class LargeStringBuilder
    /// \brief Builder class for large UTF8 strings
    class TURBO_EXPORT LargeStringBuilder : public LargeBinaryBuilder {
    public:
        using LargeBinaryBuilder::LargeBinaryBuilder;

        /// \cond FALSE
        using ArrayBuilder::finish;
        /// \endcond

        turbo::Status finish(std::shared_ptr<LargeStringArray> *out) { return finish_typed(out); }

        std::shared_ptr<DataType> type() const override { return large_utf8(); }
    };

    // ----------------------------------------------------------------------
    // BinaryViewBuilder, StringViewBuilder
    //
    // These builders do not support building raw pointer view arrays.

    namespace internal {

        // We allocate medium-sized memory chunks and accumulate data in those, which
        // may result in some waste if there are many large-ish strings. If a string
        // comes along that does not fit into a block, we allocate a new block and
        // write into that.
        //
        // Later we can implement optimizations to continuing filling underfull blocks
        // after encountering a large string that required allocating a new block.
        class TURBO_EXPORT StringHeapBuilder {
        public:
            static constexpr int64_t kDefaultBlocksize = 32 << 10;  // 32KB

            StringHeapBuilder(MemoryPool *pool, int64_t alignment)
                    : pool_(pool), alignment_(alignment) {}

            void SetBlockSize(int64_t blocksize) { blocksize_ = blocksize; }

            using c_type = BinaryViewType::c_type;

            template<bool Safe>
            std::conditional_t<Safe, turbo::Result<c_type>, c_type> append(const uint8_t *value,
                                                                           int64_t length) {
                if (length <= BinaryViewType::kInlineSize) {
                    return util::ToInlineBinaryView(value, static_cast<int32_t>(length));
                }

                if constexpr (Safe) {
                    TURBO_RETURN_NOT_OK(Reserve(length));
                }

                auto v =
                        util::ToBinaryView(value, static_cast<int32_t>(length),
                                           static_cast<int32_t>(blocks_.size() - 1), current_offset_);

                memcpy(current_out_buffer_, value, static_cast<size_t>(length));
                current_out_buffer_ += length;
                current_remaining_bytes_ -= length;
                current_offset_ += static_cast<int32_t>(length);
                return v;
            }

            static constexpr int64_t ValueSizeLimit() {
                return std::numeric_limits<int32_t>::max();
            }

            /// \brief Ensure that the indicated number of bytes can be appended via
            /// unsafe_append operations without the need to allocate more memory
            turbo::Status Reserve(int64_t num_bytes) {
                if (TURBO_UNLIKELY(num_bytes > ValueSizeLimit())) {
                    return turbo::resource_exhausted_error(
                            "BinaryView or StringView elements cannot reference "
                            "strings larger than 2GB");
                }
                if (num_bytes > current_remaining_bytes_) {
                    TURBO_RETURN_NOT_OK(FinishLastBlock());
                    current_remaining_bytes_ = num_bytes > blocksize_ ? num_bytes : blocksize_;
                    TURBO_MOVE_OR_RAISE(
                            std::shared_ptr<ResizableBuffer> new_block,
                            allocate_resizable_buffer(current_remaining_bytes_, alignment_, pool_));
                    current_offset_ = 0;
                    current_out_buffer_ = new_block->mutable_data();
                    blocks_.emplace_back(std::move(new_block));
                }
                return turbo::OkStatus();
            }

            void reset() {
                current_offset_ = 0;
                current_out_buffer_ = nullptr;
                current_remaining_bytes_ = 0;
                blocks_.clear();
            }

            int64_t current_remaining_bytes() const { return current_remaining_bytes_; }

            turbo::Result<std::vector<std::shared_ptr<ResizableBuffer>>> finish() {
                if (!blocks_.empty()) {
                    TURBO_RETURN_NOT_OK(FinishLastBlock());
                }
                current_offset_ = 0;
                current_out_buffer_ = nullptr;
                current_remaining_bytes_ = 0;
                return std::move(blocks_);
            }

        private:
            turbo::Status FinishLastBlock() {
                if (current_remaining_bytes_ > 0) {
                    // Avoid leaking uninitialized bytes from the allocator
                    TURBO_RETURN_NOT_OK(
                            blocks_.back()->resize(blocks_.back()->size() - current_remaining_bytes_,
                                    /*shrink_to_fit=*/true));
                    blocks_.back()->zero_padding();
                }
                return turbo::OkStatus();
            }

            MemoryPool *pool_;
            int64_t alignment_;
            int64_t blocksize_ = kDefaultBlocksize;
            std::vector<std::shared_ptr<ResizableBuffer>> blocks_;

            int32_t current_offset_ = 0;
            uint8_t *current_out_buffer_ = nullptr;
            int64_t current_remaining_bytes_ = 0;
        };

    }  // namespace internal

    class TURBO_EXPORT BinaryViewBuilder : public ArrayBuilder {
    public:
        using TypeClass = BinaryViewType;

        // this constructor provided for MakeBuilder compatibility
        BinaryViewBuilder(const std::shared_ptr<DataType> &, MemoryPool *pool);

        explicit BinaryViewBuilder(MemoryPool *pool = default_memory_pool(),
                                   int64_t alignment = kDefaultBufferAlignment)
                : ArrayBuilder(pool, alignment),
                  data_builder_(pool, alignment),
                  data_heap_builder_(pool, alignment) {}

        /// Set the size for future preallocated data buffers.
        ///
        /// The default size is 32KB, so after each 32KB of string data appended to the builder
        /// a new data buffer will be allocated. Adjust this to a larger value to decrease the
        /// frequency of allocation, or to a smaller value to lower the overhead of each
        /// allocation.
        void SetBlockSize(int64_t blocksize) { data_heap_builder_.SetBlockSize(blocksize); }

        /// The number of bytes which can be appended to this builder without allocating another
        /// data buffer.
        int64_t current_block_bytes_remaining() const {
            return data_heap_builder_.current_remaining_bytes();
        }

        turbo::Status append(const uint8_t *value, int64_t length) {
            TURBO_RETURN_NOT_OK(Reserve(1));
            unsafe_append_to_bitmap(true);
            TURBO_MOVE_OR_RAISE(auto v,
                                data_heap_builder_.append</*Safe=*/true>(value, length));
            data_builder_.unsafe_append(v);
            return turbo::OkStatus();
        }

        turbo::Status append(const char *value, int64_t length) {
            return append(reinterpret_cast<const uint8_t *>(value), length);
        }

        turbo::Status append(std::string_view value) {
            return append(value.data(), static_cast<int64_t>(value.size()));
        }

        /// \brief append without checking capacity
        ///
        /// Builder should have been presized using Reserve() and reserve_data(),
        /// respectively, and the value must not be larger than 2GB
        void unsafe_append(const uint8_t *value, int64_t length) {
            unsafe_append_to_bitmap(true);
            auto v = data_heap_builder_.append</*Safe=*/false>(value, length);
            data_builder_.unsafe_append(v);
        }

        void unsafe_append(const char *value, int64_t length) {
            unsafe_append(reinterpret_cast<const uint8_t *>(value), length);
        }

        void unsafe_append(const std::string &value) {
            unsafe_append(value.c_str(), static_cast<int64_t>(value.size()));
        }

        void unsafe_append(std::string_view value) {
            unsafe_append(value.data(), static_cast<int64_t>(value.size()));
        }

        /// \brief Ensures there is enough allocated available capacity in the
        /// out-of-line data heap to append the indicated number of bytes without
        /// additional allocations
        turbo::Status reserve_data(int64_t length);

        turbo::Status append_nulls(int64_t length) final {
            TURBO_RETURN_NOT_OK(Reserve(length));
            data_builder_.unsafe_append(length, BinaryViewType::c_type{});
            unsafe_set_null(length);
            return turbo::OkStatus();
        }

        /// \brief append a single null element
        turbo::Status append_null() final {
            TURBO_RETURN_NOT_OK(Reserve(1));
            data_builder_.unsafe_append(BinaryViewType::c_type{});
            unsafe_append_to_bitmap(false);
            return turbo::OkStatus();
        }

        /// \brief append a empty element (length-0 inline string)
        turbo::Status append_empty_value() final {
            TURBO_RETURN_NOT_OK(Reserve(1));
            data_builder_.unsafe_append(BinaryViewType::c_type{});
            unsafe_append_to_bitmap(true);
            return turbo::OkStatus();
        }

        /// \brief append several empty elements
        turbo::Status append_empty_values(int64_t length) final {
            TURBO_RETURN_NOT_OK(Reserve(length));
            data_builder_.unsafe_append(length, BinaryViewType::c_type{});
            unsafe_set_not_null(length);
            return turbo::OkStatus();
        }

        void unsafe_append_null() {
            data_builder_.unsafe_append(BinaryViewType::c_type{});
            unsafe_append_to_bitmap(false);
        }

        void unsafe_append_empty_value() {
            data_builder_.unsafe_append(BinaryViewType::c_type{});
            unsafe_append_to_bitmap(true);
        }

        /// \brief append a slice of a BinaryViewArray passed as an ArraySpan. Copies
        /// the underlying out-of-line string memory to avoid memory lifetime issues
        turbo::Status append_array_slice(const ArraySpan &array, int64_t offset,
                                       int64_t length) override;

        void reset() override;

        turbo::Status resize(int64_t capacity) override {
            TURBO_RETURN_NOT_OK(check_capacity(capacity));
            capacity = std::max(capacity, kMinBuilderCapacity);
            TURBO_RETURN_NOT_OK(data_builder_.resize(capacity));
            return ArrayBuilder::resize(capacity);
        }

        turbo::Status finish_internal(std::shared_ptr<ArrayData> *out) override;

        std::shared_ptr<DataType> type() const override { return binary_view(); }

    protected:
        TypedBufferBuilder<BinaryViewType::c_type> data_builder_;

        // Accumulates out-of-line data in fixed-size chunks which are then attached
        // to the resulting ArrayData
        internal::StringHeapBuilder data_heap_builder_;
    };

    class TURBO_EXPORT StringViewBuilder : public BinaryViewBuilder {
    public:
        using BinaryViewBuilder::BinaryViewBuilder;

        std::shared_ptr<DataType> type() const override { return utf8_view(); }
    };

    // ----------------------------------------------------------------------
    // FixedSizeBinaryBuilder

    class TURBO_EXPORT FixedSizeBinaryBuilder : public ArrayBuilder {
    public:
        using TypeClass = FixedSizeBinaryType;

        explicit FixedSizeBinaryBuilder(const std::shared_ptr<DataType> &type,
                                        MemoryPool *pool = default_memory_pool(),
                                        int64_t alignment = kDefaultBufferAlignment);

        turbo::Status append(const uint8_t *value) {
            TURBO_RETURN_NOT_OK(Reserve(1));
            unsafe_append(value);
            return turbo::OkStatus();
        }

        turbo::Status append(const char *value) {
            return append(reinterpret_cast<const uint8_t *>(value));
        }

        turbo::Status append(std::string_view view) {
            TURBO_RETURN_NOT_OK(Reserve(1));
            unsafe_append(view);
            return turbo::OkStatus();
        }

        turbo::Status append(const std::string &s) {
            TURBO_RETURN_NOT_OK(Reserve(1));
            unsafe_append(s);
            return turbo::OkStatus();
        }

        turbo::Status append(BufferSpan buf) {
            TURBO_RETURN_NOT_OK(Reserve(1));
            unsafe_append(buf);
            return turbo::OkStatus();
        }

        turbo::Status append(const std::shared_ptr<Buffer> &s) { return append(*s); }

        template<size_t NBYTES>
        turbo::Status append(const std::array<uint8_t, NBYTES> &value) {
            TURBO_RETURN_NOT_OK(Reserve(1));
            unsafe_append(
                    std::string_view(reinterpret_cast<const char *>(value.data()), value.size()));
            return turbo::OkStatus();
        }

        turbo::Status append_values(const uint8_t *data, int64_t length,
                                   const uint8_t *valid_bytes = nullptr);

        turbo::Status append_values(const uint8_t *data, int64_t length, const uint8_t *validity,
                                   int64_t bitmap_offset);

        turbo::Status append_null() final;

        turbo::Status append_nulls(int64_t length) final;

        turbo::Status append_empty_value() final;

        turbo::Status append_empty_values(int64_t length) final;

        turbo::Status append_array_slice(const ArraySpan &array, int64_t offset,
                                       int64_t length) override {
            return append_values(
                    array.get_values<uint8_t>(1, 0) + ((array.offset + offset) * byte_width_), length,
                    array.get_values<uint8_t>(0, 0), array.offset + offset);
        }

        void unsafe_append(const uint8_t *value) {
            unsafe_append_to_bitmap(true);
            if (TURBO_LIKELY(byte_width_ > 0)) {
                byte_builder_.unsafe_append(value, byte_width_);
            }
        }

        void unsafe_append(const char *value) {
            unsafe_append(reinterpret_cast<const uint8_t *>(value));
        }

        void unsafe_append(std::string_view value) {
#ifndef NDEBUG
            check_value_size(static_cast<size_t>(value.size()));
#endif
            unsafe_append(reinterpret_cast<const uint8_t *>(value.data()));
        }

        void unsafe_append(BufferSpan s) {
#ifndef NDEBUG
            check_value_size(s.size());
#endif
            unsafe_append(s.data());
        }

        void unsafe_append(const std::shared_ptr<Buffer> &s) { unsafe_append((BufferSpan)*s); }

        void unsafe_append_null() {
            unsafe_append_to_bitmap(false);
            byte_builder_.unsafe_append(/*num_copies=*/byte_width_, 0);
        }

        turbo::Status validate_overflow(int64_t new_bytes) const {
            auto new_size = byte_builder_.length() + new_bytes;
            if (TURBO_UNLIKELY(new_size > memory_limit())) {
                return turbo::resource_exhausted_error("array cannot contain more than ", memory_limit(),
                                                       " bytes, have ", new_size);
            } else {
                return turbo::OkStatus();
            }
        }

        /// \brief Ensures there is enough allocated capacity to append the indicated
        /// number of bytes to the value data buffer without additional allocations
        turbo::Status reserve_data(int64_t elements) {
            TURBO_RETURN_NOT_OK(validate_overflow(elements));
            return byte_builder_.Reserve(elements);
        }

        void reset() override;

        turbo::Status resize(int64_t capacity) override;

        turbo::Status finish_internal(std::shared_ptr<ArrayData> *out) override;

        /// \cond FALSE
        using ArrayBuilder::finish;
        /// \endcond

        turbo::Status finish(std::shared_ptr<FixedSizeBinaryArray> *out) { return finish_typed(out); }

        /// \return size of values buffer so far
        int64_t value_data_length() const { return byte_builder_.length(); }

        int32_t byte_width() const { return byte_width_; }

        /// Temporary access to a value.
        ///
        /// This pointer becomes invalid on the next modifying operation.
        const uint8_t *get_value(int64_t i) const;

        /// Temporary access to a value.
        ///
        /// This view becomes invalid on the next modifying operation.
        std::string_view get_view(int64_t i) const;

        static constexpr int64_t memory_limit() {
            return std::numeric_limits<int64_t>::max() - 1;
        }

        std::shared_ptr<DataType> type() const override {
            return fixed_size_binary(byte_width_);
        }

    protected:
        int32_t byte_width_;
        BufferBuilder byte_builder_;

        /// Temporary access to a value.
        ///
        /// This pointer becomes invalid on the next modifying operation.
        uint8_t *GetMutableValue(int64_t i) {
            uint8_t *data_ptr = byte_builder_.mutable_data();
            return data_ptr + i * byte_width_;
        }

        void check_value_size(int64_t size);
    };

    /// @}

    // ----------------------------------------------------------------------
    // Chunked builders: build a sequence of BinaryArray or StringArray that are
    // limited to a particular size (to the upper limit of 2GB)

    namespace internal {

        class TURBO_EXPORT ChunkedBinaryBuilder {
        public:
            explicit ChunkedBinaryBuilder(int32_t max_chunk_value_length,
                                          MemoryPool *pool = default_memory_pool());

            ChunkedBinaryBuilder(int32_t max_chunk_value_length, int32_t max_chunk_length,
                                 MemoryPool *pool = default_memory_pool());

            virtual ~ChunkedBinaryBuilder() = default;

            turbo::Status append(const uint8_t *value, int32_t length) {
                if (TURBO_UNLIKELY(length + builder_->value_data_length() >
                                   max_chunk_value_length_)) {
                    if (builder_->value_data_length() == 0) {
                        // The current item is larger than max_chunk_size_;
                        // this chunk will be oversize and hold *only* this item
                        TURBO_RETURN_NOT_OK(builder_->append(value, length));
                        return NextChunk();
                    }
                    // The current item would cause builder_->value_data_length() to exceed
                    // max_chunk_size_, so finish this chunk and append the current item to the next
                    // chunk
                    TURBO_RETURN_NOT_OK(NextChunk());
                    return append(value, length);
                }

                if (TURBO_UNLIKELY(builder_->length() == max_chunk_length_)) {
                    // The current item would cause builder_->length() to exceed max_chunk_length_, so
                    // finish this chunk and append the current item to the next chunk
                    TURBO_RETURN_NOT_OK(NextChunk());
                }

                return builder_->append(value, length);
            }

            turbo::Status append(std::string_view value) {
                return append(reinterpret_cast<const uint8_t *>(value.data()),
                              static_cast<int32_t>(value.size()));
            }

            turbo::Status append_null() {
                if (TURBO_UNLIKELY(builder_->length() == max_chunk_length_)) {
                    TURBO_RETURN_NOT_OK(NextChunk());
                }
                return builder_->append_null();
            }

            turbo::Status Reserve(int64_t values);

            virtual turbo::Status finish(ArrayVector *out);

        protected:
            turbo::Status NextChunk();

            // maximum total character data size per chunk
            int64_t max_chunk_value_length_;

            // maximum elements allowed per chunk
            int64_t max_chunk_length_ = kListMaximumElements;

            // when Reserve() would cause builder_ to exceed its max_chunk_length_,
            // add to extra_capacity_ instead and wait to reserve until the next chunk
            int64_t extra_capacity_ = 0;

            std::unique_ptr<BinaryBuilder> builder_;
            std::vector<std::shared_ptr<Array>> chunks_;
        };

        class TURBO_EXPORT ChunkedStringBuilder : public ChunkedBinaryBuilder {
        public:
            using ChunkedBinaryBuilder::ChunkedBinaryBuilder;

            turbo::Status finish(ArrayVector *out) override;
        };

    }  // namespace internal

}  // namespace nebula
